#include <sstream>
#include <time.h>
#include "EventBus.h"
#include "Config.h"

using namespace etb;

EventBus::EventBus()
  : running_(false)
{
}

EventBus::~EventBus()
{
}

void EventClient::parseEvent(const std::string event, EventStruct* evts)
{
  if (evts == NULL)
    return;

  evts->target = "";
  evts->device = "";
  evts->param = "";

  const char *delim = " ";
  std::vector<std::string> v;

  size_t s;
  s = event.find_first_of(delim);
  v.push_back(event.substr(0, s));

  while (1) {
    size_t p = s;
    s = event.find_first_of(delim, p + 1);
    if ( (s == std::string::npos) && (p + 1 != event.length()) ) {
      v.push_back(event.substr(p + 1));
      break;
    }

    if ( (s != p) && (v.size() < 4) )
      v.push_back(event.substr(p + 1, s - p - 1));
  }

  evts->target = v[0];
  if (v.size() > 1)
    evts->device = v[1];
  if (v.size() > 2)
    evts->param = v[2];
}

std::string EventClient::makeCmd(const std::string target, 
                                 const std::string device, 
                                 const std::string param)
{
  std::ostringstream oss(std::ostringstream::out);
  oss << target << " " << device << " " << param;

  return oss.str();
}

void EventBus::registerEventClient(EventClient* client)
{
  std::pair<std::set<EventClient*>::iterator, bool> ret;
  ret = clients_.insert(client);

  if (ret.second == false) {
    MSG("[EVENTBUS] Registering duplicate client.");
  }
}

void EventBus::unregisterEventClient(EventClient* client)
{
  clients_.erase(client);
}

void EventBus::start()
{
  int ret = 0;
  
  ret = pthread_mutex_init(&mutex_, NULL);
  ASSERT(ret == 0);

  ret = pthread_cond_init(&cond_, NULL);
  ASSERT(ret == 0);

  ret = pthread_create(&thread_, NULL, EventBus::threadEntry, this);
  ASSERT(ret == 0);
  running_ = true;
}

void EventBus::stop()
{
  running_ = false;
  pthread_join(thread_, NULL);
}

void* EventBus::threadEntry(void* arg)
{
  if (arg) 
    static_cast<EventBus*>(arg)->threadProc();

  return NULL;
}

void EventBus::threadProc()
{
  MSG("Thread started.");
  while(1) {
    std::string evt;

    if (running_ == false) break;

    pthread_mutex_lock(&mutex_);

    struct timespec ts;
    makeTimespec(&ts, 0, 10000);

    pthread_cond_timedwait(&cond_, &mutex_, &ts);
    if (!queue_.empty()) {
      evt = queue_.front();
      queue_.pop_front();
    }

    pthread_mutex_unlock(&mutex_);

    if (!evt.empty())
      callbackAllClients(evt);
  }

  MSG("Thread stopped.");
}

void EventBus::sendEvent(const std::string event)
{
  std::cout << "[BUS] Send: " << event << std::endl;
  pthread_mutex_lock(&mutex_);

  queue_.push_back(event);
  pthread_cond_signal(&cond_);

  pthread_mutex_unlock(&mutex_);
}

void EventBus::callbackAllClients(const std::string event)
{
  std::set<EventClient*>::const_iterator i;
  for (i = clients_.begin(); i != clients_.end(); i++) {
    if (*i) {
      dynamic_cast<EventClient*>(*i)->handleEvent(event);
    }
  }
}

void EventBus::makeTimespec(struct timespec* ts, int sec, int nsec)
{
  if (ts) {
    ts->tv_sec = time(0) + sec;
    ts->tv_nsec = nsec;
  }
}
